package com.hbase.zhou.db;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.util.List;

/**
 * @Description:
 * @Author: ZhOu
 * @Date: 2018/3/20
 */
public class HBaseUtil {

    private static Configuration configuration;
    private static Connection connection;

    static {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "node1,node2,node3");

        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取连接
     */
    public static Connection getConnection() {
        if (connection == null || connection.isClosed()) {
            try {
                connection = ConnectionFactory.createConnection(configuration);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return connection;
    }

    /**
     * 关闭连接
     */
    public static void close() {
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 新建表
     *
     * @param tableName          表名
     * @param hColumnDescriptors 列簇列表
     * @return 创建结果
     */
    public static boolean createTable(String tableName, List<HColumnDescriptor> hColumnDescriptors) {
        HBaseAdmin hBaseAdmin = null;
        TableName tbName = TableName.valueOf(tableName);
        try {
            hBaseAdmin = (HBaseAdmin) connection.getAdmin();

            if (dropTable(tableName)) {
                HTableDescriptor descriptor = new HTableDescriptor(tbName);
                //设置日志级别
                descriptor.setDurability(Durability.SYNC_WAL);

                //设置memstore内存大小，1M
                descriptor.setMemStoreFlushSize(1048576);

                //设置region中store文件的最大size,这里设为50M
                descriptor.setMaxFileSize(52428800);

                hColumnDescriptors.forEach(column -> descriptor.addFamily(column));
                hBaseAdmin.createTable(descriptor);
            }
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (hBaseAdmin != null) {
                try {
                    hBaseAdmin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            HBaseUtil.close();
        }
        return false;
    }

    /**
     * 删除表
     *
     * @param tableName 表名
     */
    public static boolean dropTable(String tableName) {
        TableName tbName = TableName.valueOf(tableName);
        try {
            HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
            if (admin.tableExists(tbName)) {
                admin.disableTable(tbName);
                admin.deleteTable(tableName);
            }
            admin.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 插入数据
     *
     * @param tableName 表名
     * @param rowKey    主键
     * @param family    列簇
     * @param qualifier 列
     * @param value     值
     * @return 是否插入成功
     */
    public static boolean putData(String tableName, String rowKey, String family,
                                  String qualifier, String value) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(rowKey.getBytes());
            put.addColumn(family.getBytes(), qualifier.getBytes(), value.getBytes());
            table.put(put);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 批量插入数据
     *
     * @param tableName 表名
     * @param puts      数据集合
     * @return 执行结果
     */
    public static boolean putData(String tableName, List<Put> puts) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            table.put(puts);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @return
     */
    public static boolean deleteData(String tableName, String rowKey,
                                     String family, String qualifier) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(rowKey.getBytes());
            delete.addColumn(family.getBytes(), qualifier.getBytes());

        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
}
